package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/1
 *
 *  检测每种传感器的水位值，如果连续的两个水位值超过10，就输出报警
 */
public class Demo3_ValueState
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .process(new KeyedProcessFunction<String, WaterSensor, String>()
                   {
                       //存储一个变量，保存上一个传感器的水位
                       private ValueState<Integer> lastVc;

                       @Override
                       public void open(Configuration parameters) throws Exception {
                           //初始化状态
                           lastVc = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVc", Integer.class));
                       }


                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                            //如果连续的两个水位值超过10，就输出报警
                           if (lastVc.value() != null && value.getVc() > 10 && lastVc.value() > 10 ){
                               out.collect("预警:连续的两个水位值超过10");
                           }
                           //存入当前水位
                           lastVc.update(value.getVc());

                       }
                   })
                   .print();
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
    }
}
